package com.atguigu.gmall.realtime.app.dwd;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONAware;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.app.BaseAppV1;
import com.atguigu.gmall.realtime.common.Constant;
import com.atguigu.gmall.realtime.util.AtguiguUtil;
import com.atguigu.gmall.realtime.util.FlinkSinkUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.time.Duration;
import java.util.Comparator;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * @author Archie
 * @date 2021-10-17 15:02
 * @description
 */
public class DwdLogApp extends BaseAppV1 {

	public static final String PAGE = "page";
	public static final String START = "start";
	public static final String DISPLAY = "display";

	public static void main(String[] args) {
		new DwdLogApp().init(2001, 1, "DwdLogApp", "DwdLogApp", Constant.TOPIC_ODS_LOG);

	}

	@Override
	protected void run(StreamExecutionEnvironment env,
					   DataStreamSource<String> stream) {
		// 1. 区别新老客户（给 is_new 字段重新赋值）
		SingleOutputStreamOperator<JSONObject> validatedStream = distinguishNewOrOld(stream);
		// 2. 数据分流
		Map<String, DataStream<JSONObject>> splitStreams = splitStream(validatedStream);
		// 3. 不同数据写入不同Topic
		writeToKafka(splitStreams);

	}

	/**
	 * 将数据写入Kafka中
	 * @param splitStreams
	 */
	private void writeToKafka(Map<String, DataStream<JSONObject>> splitStreams) {
		splitStreams.get(START)
				.map(JSONAware::toJSONString)
				.addSink(FlinkSinkUtil.getKafkaSink(Constant.TOPIC_DWD_START));
		splitStreams.get(PAGE)
				.map(JSONAware::toJSONString)
				.addSink(FlinkSinkUtil.getKafkaSink(Constant.TOPIC_DWD_PAGE));
		splitStreams.get(DISPLAY)
				.map(JSONAware::toJSONString)
				.addSink(FlinkSinkUtil.getKafkaSink(Constant.TOPIC_DWD_DISPLAY));
	}


	/**
	 * 分流
	 * @param validatedStream 输入数据
	 * @return
	 */
	private Map<String, DataStream<JSONObject>> splitStream(SingleOutputStreamOperator<JSONObject> validatedStream) {

		OutputTag<JSONObject> pageTag = new OutputTag<JSONObject>("page") {};
		OutputTag<JSONObject> displayTag = new OutputTag<JSONObject>("display") {};
		Map<String, DataStream<JSONObject>> result = new HashMap<>();

		SingleOutputStreamOperator<JSONObject> pageStream = validatedStream.process(new ProcessFunction<JSONObject, JSONObject>() {
			@Override
			public void processElement(JSONObject jsonObject, Context context, Collector<JSONObject> collector) throws Exception {
				// 【主流】启动日志
				// 【测流】页面日志 & 曝光日志
				JSONObject startObject = jsonObject.getJSONObject("start");
				if (startObject != null) {
					/* 启动日志处理 */
					collector.collect(jsonObject);
				} else {
					/* 页面日志处理 */
					JSONObject pageObject = jsonObject.getJSONObject("page");
					if (pageObject != null) {
						context.output(pageTag, jsonObject);
					}
					/* 曝光日志处理（为了后续方便，直接拆开这个数组遍历输出） */
					JSONArray displays = jsonObject.getJSONArray("displays");
					if (displays != null) {
						for (int i = 0; i < displays.size(); i++) {
							JSONObject display = displays.getJSONObject(i);
							/* 补上外层的其他数据 */
							display.putAll(jsonObject.getJSONObject("common"));
							display.putAll(jsonObject.getJSONObject("page"));
							display.put("ts", jsonObject.getLong("ts"));
							context.output(displayTag, display);
						}
					}
				}
			}
		});

		result.put(START, pageStream);
		result.put(PAGE, pageStream.getSideOutput(pageTag));
		result.put(DISPLAY, pageStream.getSideOutput(displayTag));
		return result;
	}

	/**
	 * 区别新老客户
	 * 新客户 - 首次访问的客户
	 * @param stream 输入数据
	 * @return
	 */
	private SingleOutputStreamOperator<JSONObject> distinguishNewOrOld(DataStreamSource<String> stream) {
		// 找到首次访问记录（使用状态标识是否首次访问）+（事件时间&窗口）
		// 【问题】如何判断第一个窗口？
		// 【解决】时间戳最小的 is_new => 1 其它的 is_new => 0
		return stream
				.map(JSON::parseObject)
				/* 设置时间戳 & 水印 */
				.assignTimestampsAndWatermarks(
						WatermarkStrategy
								.<JSONObject>forBoundedOutOfOrderness(Duration.ofSeconds(3))
						.withTimestampAssigner((obj, ts) -> obj.getLong("ts"))
				)
				/* 分组 */
				.keyBy(obj -> obj.getJSONObject("common").getString("mid"))
				/* 开滑动窗口 */
				.window(TumblingEventTimeWindows.of(Time.seconds(5)))
				/* 处理 */
				.process(new ProcessWindowFunction<JSONObject, JSONObject, String, TimeWindow>() {
					private ValueState<Boolean> isFirstWindowState;

					@Override
					public void open(Configuration parameters) throws Exception {
						isFirstWindowState = getRuntimeContext().getState(new ValueStateDescriptor<Boolean>(
								"isFirstWindowState",
								Boolean.class
						));
					}

					@Override
					public void process(String s, Context context, Iterable<JSONObject> iterable, Collector<JSONObject> collector) throws Exception {
						// 找到第一个窗口，按照时间戳排序，最小的设置 is_new为1
						if (isFirstWindowState.value() == null) { /* 第一个窗口 */
							isFirstWindowState.update(true);
							List<JSONObject> list = AtguiguUtil.toList(iterable);
							list.sort(Comparator.comparing(o -> o.getLong("ts")));
							for (int i = 0; i < list.size(); i++) {
								JSONObject obj = list.get(i);
								if (i == 0) {
									obj.getJSONObject("common").put("is_new", "1");
								} else {
									obj.getJSONObject("common").put("is_new", "0");
								}
								collector.collect(obj);
							}

						} else { /* 非第一个窗口 */
							for (JSONObject object : iterable) {
								object.getJSONObject("common").put("mid", "0");
								collector.collect(object);
							}
						}

					}
				});



	}
}
